package zisu.rabbit.rabbitmqspring;

import com.rabbitmq.client.Channel;
import org.apache.logging.log4j.message.SimpleMessageFactory;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionListener;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.support.ConsumerTagStrategy;
import org.springframework.amqp.support.converter.ContentTypeDelegatingMessageConverter;
import org.springframework.amqp.support.converter.DefaultJackson2JavaTypeMapper;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import zisu.rabbit.rabbitmqspring.convert.ImageMessageConverter;
import zisu.rabbit.rabbitmqspring.convert.PDFMessageConvert;
import zisu.rabbit.rabbitmqspring.convert.TextMessageConvert;
import zisu.rabbit.rabbitmqspring.entity.Order;
import zisu.rabbit.rabbitmqspring.entity.Packaged;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

/**
 * @Author: 蔡翔
 * @Date: 2019/7/25 13:49
 * @Version 1.0
 */
@Configuration
@ComponentScan("zisu.rabbit.*")
public class RabbitMQConfig{
    @Bean //把这个Bean(这个connectionFactory )。已近注入到Spring 容器了
    public CachingConnectionFactory connectionFactory(){
        /*ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.0.229");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setVirtualHost("/");*/
        CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
        cachingConnectionFactory.setVirtualHost("/");
        cachingConnectionFactory.setUsername("guest");
        cachingConnectionFactory.setAddresses("192.168.0.151:5672");
        cachingConnectionFactory.setPassword("guest");
        return cachingConnectionFactory;
    }

    //把RabbitAdmin 也注入到Spring 容器里面去。 在其他Controller层只要注入进去就可以使用了。
    @Bean
    public RabbitAdmin rabbitAdmin(CachingConnectionFactory connectionFactory) throws Exception{
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        rabbitAdmin.setAutoStartup(true);
        return rabbitAdmin;
    }

    @Bean
    public TopicExchange exchange001(){
        return new TopicExchange("topic001",true,true);
    }

    @Bean
    public Queue queue001(){
        return new Queue("queue001",true);
    }

    @Bean
    public Binding binding001(){
        return BindingBuilder.bind(queue001()).to(exchange001()).with("spring.*");
    }

    @Bean
    public TopicExchange exchange002(){
        return new TopicExchange("topic002",true,true);
    }

    @Bean
    public Queue queue002(){
        return new Queue("queue002",true);
    }

    @Bean
    public Binding binding002(){
        return BindingBuilder.bind(queue002()).to(exchange002()).with("rabbit.*");
    }

    @Bean
    public Queue queue003(){
        return new Queue("queue003",true);
    }

    @Bean
    public Binding binding003(){
        return BindingBuilder.bind(queue003()).to(exchange001()).with("mq.*");
    }

    @Bean
    public Queue queue_image(){
        //队列持久
        return new Queue("image_queue",true);
    }

    @Bean
    public Queue queue_pdf(){
        //队列持久
        return new Queue("pdf_queue",true);
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory){
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        return rabbitTemplate;
    }

    // 这个监听器必须加上Bean注解，让spring 在启动的时候就加入到容器中，它是监听队列的，和业务send什么的没关系。
    // 把某种格式的消息，指定对应特定的convert转换器，然后用全局转换器来委派到这个转换器里面去执行特定的方法然后return给 MessageDelegate,去执行里面特定的方法。
    @Bean
    public SimpleMessageListenerContainer messageListenerContainer(ConnectionFactory connectionFactory){
        SimpleMessageListenerContainer simpleMessageListenerContainer = new SimpleMessageListenerContainer(connectionFactory);
        simpleMessageListenerContainer.setQueues(queue001(),queue002(),queue003(),queue_image(),queue_pdf()); //设置你要监听的队列。
        simpleMessageListenerContainer.setConcurrentConsumers(1);  //指定当前迸发消费者数量
        ////注意： 这里的消费者是监听容器的，他们一个消费者会去监听一个队列
        simpleMessageListenerContainer.setMaxConcurrentConsumers(5);  //最大迸发消费者数量为5个
        simpleMessageListenerContainer.setDefaultRequeueRejected(false);  //设置默认重回队列为false
        simpleMessageListenerContainer.setAcknowledgeMode(AcknowledgeMode.AUTO); //设置签收模式
        simpleMessageListenerContainer.setConsumerTagStrategy(new ConsumerTagStrategy() {
            @Override
            public String createConsumerTag(String queue) {
                return queue+"_"+ UUID.randomUUID().toString(); //给消费者自定义 设置一个标签
            }
        }); //设置消费端的标签策略

        //1.
        /*//用适配器 去取代 监听器（listener）。
        simpleMessageListenerContainer.setMessageListener(new ChannelAwareMessageListener() {
            //***ChannelAwareMessageListener 是匿名的消息接口，如果有消息过来，就会通过这个类进行监听，然后走到onMessage这个方法
            @Override
            public void onMessage(Message message, Channel channel) throws Exception {
                String msg = new String(message.getBody());
                System.out.println("----------监听到的消息："+msg);
            }
        });*/
        // 有消息进入 适配器后，它会自动执行MessageDelegate类 里面的bandleMessage 方法。

        /*//2.
        // 监听队列的流程是这样的：1.先监听到队列里面的数据然后调用messageConverter的方法转成string，最后调用MessageDelegate类里面的 方法，形参是string。
        MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate()); //通过适配器，去适配不同的实现。
        adapter.setDefaultListenerMethod("consumeMessage");
        adapter.setMessageConverter(new TextMessageConvert()); //这里设置了messageConvert，在就听到这条消息的时候，会把这条消息里面的body 转化成string。
        simpleMessageListenerContainer.setMessageListener(adapter);
        return simpleMessageListenerContainer;*/

        //3. 适配器方式：我们的队列名称和方法名称 也可以进行 一一对应
        /*MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate()); //不管怎么样 都需要一个adapter 和一个委派对象（new MessageDelegate()）
        adapter.setMessageConverter(new TextMessageConvert()); //因为委派方法里面需要的是String类型的参数，所以我们这里需要setMessageConverter一下。
        simpleMessageListenerContainer.setMessageListener(adapter);
        Map<String,String> queueOrTagToMethodName = new HashMap<>();
        queueOrTagToMethodName.put("queue001","method1");
        queueOrTagToMethodName.put("queue002","method2");
        adapter.setQueueOrTagToMethodName(queueOrTagToMethodName);*/

        /*//1.1 支持json格式的转换器
        MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
        adapter.setDefaultListenerMethod("consumeMessage");

        Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
        adapter.setMessageConverter(jackson2JsonMessageConverter);
        simpleMessageListenerContainer.setMessageListener(adapter);*/


        /*//1.2 支持java对象转换
        MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
        adapter.setDefaultListenerMethod("consumeMessage");

        Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();

        //就这个类，可以吧json格式转换成 java对象。只要在消息头上加上 “__TypeId__” ，就能被DefaultJackson2JavaTypeMapper识别。
        DefaultJackson2JavaTypeMapper defaultJackson2JavaTypeMapper = new DefaultJackson2JavaTypeMapper();
        // * 这个类主要是为了解决：If you believe this class is safe to deserialize, please provide its name. If the serialization is only done by a trusted source, you can also enable trust all (*).
        defaultJackson2JavaTypeMapper.setTrustedPackages("*");

        jackson2JsonMessageConverter.setJavaTypeMapper(defaultJackson2JavaTypeMapper);

        adapter.setMessageConverter(jackson2JsonMessageConverter);
        simpleMessageListenerContainer.setMessageListener(adapter);*/

        /*//1.3 Default 支持java对象多映射转换
        MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
        adapter.setDefaultListenerMethod("consumeMessage");

        Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
        DefaultJackson2JavaTypeMapper javaTypeMapper = new DefaultJackson2JavaTypeMapper();

        Map<String,Class<?>> idClassMapping = new HashMap<String, Class<?>>();
        idClassMapping.put("order", Order.class);
        idClassMapping.put("packaged", Packaged.class);

        javaTypeMapper.setIdClassMapping(idClassMapping);
        jackson2JsonMessageConverter.setJavaTypeMapper(javaTypeMapper);
        adapter.setMessageConverter(jackson2JsonMessageConverter);
        simpleMessageListenerContainer.setMessageListener(adapter);*/

        //1.4 ext convert ；；指定转换器转换后返回的数据传达到哪里（MessageDelegate类的 consumeMessage方法（依据参数））
        MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
        adapter.setDefaultListenerMethod("consumeMessage");

        //全局的转换器，这是一个很大的转换器
        ContentTypeDelegatingMessageConverter converter = new ContentTypeDelegatingMessageConverter();
        // 无论是 text 或者 html/text ，都是走TextMessageConvert()这个转换器。
        TextMessageConvert textConvert = new TextMessageConvert();
        converter.addDelegate("text",textConvert);
        converter.addDelegate("html/text",textConvert);
        converter.addDelegate("xml/text",textConvert);
        converter.addDelegate("text/plain",textConvert);


        Jackson2JsonMessageConverter jsonConvert = new Jackson2JsonMessageConverter();
        converter.addDelegate("json",jsonConvert);
        converter.addDelegate("application/json",jsonConvert);

        //ImageMessageConverter
        ImageMessageConverter imageMessageConverter = new ImageMessageConverter();
        converter.addDelegate("image/png",imageMessageConverter);
        converter.addDelegate("image",imageMessageConverter);

        //PDFMessageConverter
        PDFMessageConvert pdfMessageConvert = new PDFMessageConvert();
        converter.addDelegate("application/pdf",pdfMessageConvert);

        adapter.setMessageConverter(converter);
        simpleMessageListenerContainer.setMessageListener(adapter);

        return simpleMessageListenerContainer;
    }

}
